package com.cn.broker.processor;

import com.cn.broker.BrokerController;
import com.cn.broker.PullRequest;
import com.cn.common.ConsumeQueueData;
import com.cn.common.PullResult;
import com.cn.common.PullResultStatus;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.file.GetMessageResult;
import com.cn.file.GetMessageStatus;
import com.cn.msg.Message;
import com.cn.util.RemotingSerializable;
import com.cn.util.RunEnv;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.omg.SendingContext.RunTime;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

/**
 * 拉取消息业务
 */
public class PullRequestProcessor implements INettyRequestProcessor {

    private BrokerController brokerController;

    public PullRequestProcessor(BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    @Override
    public Message process(Message message, ChannelHandlerContext ctx) {
        return process0(message, ctx.channel(), true);
    }

    public Message process0(Message message, Channel channel, boolean suspendHold) {
        ConsumeQueueData consumeQueueData = RemotingSerializable.decode(message.getByteBodys(), ConsumeQueueData.class);
        message.setType((byte) 11);
        message.setFlag((byte) 0);
        message.setLength(0);
        PullResult pullResult = new PullResult();
        pullResult.setStatus(PullResultStatus.NOT_FOUND);

        if (consumeQueueData != null) {
            //查询消息
            GetMessageResult result = this.brokerController.getDefaultMessageStore().getMessage(consumeQueueData.getTopic(), (int) consumeQueueData.getQueueId(), consumeQueueData.getOffset());
            if (result != null) {
                switch (result.getMessageStatus()) {
                    case FOUND:
                        pullResult.setStatus(PullResultStatus.FOUND);
                        pullResult.setBodys(getMsgBody(result));//消息体
                        pullResult.setMinOffset(result.getMinOffset());
                        pullResult.setMaxOffset(result.getMaxOffset());
                        pullResult.setNextOffset(result.getNextOffset());//下次拉取位移
                        pullResult.setTotalMsgSize(result.getTotalMsgSize());
                        break;
                    case NOT_FOUND:
                        //锁定5秒（时间可以配置）
                        if (suspendHold) {
                            PullRequest pullRequest = new PullRequest(consumeQueueData.getTopic(), (int) consumeQueueData.getQueueId(), consumeQueueData.getOffset(), System.currentTimeMillis(), channel, message);
                            this.brokerController.getPullMessageRequestHoldService().suspendRequest(pullRequest);
                            pullResult = null;
                        }
                        break;
                }
            }
        }
        if (pullResult != null) {
            byte[] bytes = RemotingSerializable.encode(pullResult);
            message.setByteBodys(bytes);
            message.setLength(bytes.length);
            return message;
        } else {
            return null;
        }
    }

    public void wakeupHoldRequest(Channel channel, Message message, PullRequest request) {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                Message response = process0(message, channel, false);
                if (response != null) {
                    response.setFlag((byte) 0);
                    try {
                        channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture ct) throws Exception {

                            }
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        this.brokerController.getDefaultExecutorService().submit(runnable);
    }

    public byte[] getMsgBody(GetMessageResult getMessageResult) {
        ByteBuffer byteBuffer = ByteBuffer.allocate((int) getMessageResult.getTotalMsgSize());
        if (getMessageResult.getMessageStatus() == GetMessageStatus.FOUND) {
            try {
                List<ByteBuffer> list = getMessageResult.getMsgList();
                for (ByteBuffer buffer : list) {
                    byteBuffer.put(buffer);
                }
            } finally {
                getMessageResult.release();
            }
        }
        return byteBuffer.array();
    }
}
